<?php
/**
 * Created by PhpStorm.
 * User: wuxiansheng
 * Date: 2018/8/18
 * Time: 19:06
 */

namespace App\Console\Service\IPProxyPool\Process;


use App\Console\Service\IPProxyPool\Inc\Rule;
use App\Console\Service\IPProxyPool\Uitl\Monolog;
use Curl\Curl;
use Illuminate\Support\Facades\Redis;

class Download
{
    /**
     * 多进程执行逻辑
     */
    public function run()
    {

        $logger = function (string $str) {
            Monolog::info('IPProxyPool----->>>>>>>>download ' . $str);
        };

        $logger('Process - beginning');
        $t1 = microtime(true);

        //创建任务
        $this->createMission($logger);

        //创建过程
        $worker_num = 1;
        $process_pool = [];
        $process = null;

        //创建worker进程
        for ($i = 0; $i < $worker_num; $i++) {
            $process = new \swoole_process(function (\swoole_process $worker) use ($logger) {
                $this->getProxyList($worker, $logger); //执行过程
            });
            $process->start();
            $pid = $process->pid;
            $process_pool[$pid] = $process;
        }

        //子进程结束必须要执行wait进行回收，否则子进程会变成僵尸进程
        while ($ret = \swoole_process::wait()) {// $ret 是个数组 code是进程退出状态码，
            $pid = $ret['pid'];
            $code = $ret['code'];
            $logger("worker Exit, PID=" . $pid . " CODE=" . $code);
        }
        $logger("Master exit");
        $t2 = microtime(true);
        $logger('耗时' . round($t2 - $t1, 3) . '秒');
        $logger('Now memory_get_usage: ' . memory_get_usage());

    }

    /**
     * @param $logger
     */
    public function createMission($logger)
    {
        //获取规则 && 创建下载列表
        $proxyList = Rule::getUrlsList();

        Redis::del("lists");
        Redis::del("index:types:wait");
        //添加队列
        try {
            foreach ($proxyList as $key => $url) {
                Redis::rpush("lists", json_encode($url));
            }
        } catch (Exception $e) {
            $logger("Message: " . $e->getMessage());
            return;
        }
    }

    /**
     * @param \swoole_process $worker
     * @param $logger
     */
    public function getProxyList(\swoole_process $worker, $logger)
    {

        $curl = new Curl;
        while ($items = Redis::lpop("lists")) {
            $item = json_decode($items, true);
            if (!empty($item['url'])) {
                $curl->get($item['url']);
                $logger('worker:' . $worker->pid . ' html:' . $item['url']);
                $body = $curl->response;
                if (!empty($body)) {
                    try {
                        $proxyItemArr = Rule::getProxy($body, $item["code"], $item["pattern"], $item["position"]);
                        foreach ($proxyItemArr as $proxyKey => $proxyItem) {
                            $key = "proxy:" . $proxyItem["ip"] . ':' . $proxyItem["port"];
                            Redis::hmset($key, $proxyItem);
                            Redis::SADD("index:types:wait", $key);
                        }
                    } catch (Exception $e) {
                        var_dump($e->getMessage());
                    }
                } else {
                    $logger('worker:' . $worker->pid . 'curl error');
                    sleep(1);
                }
            } else {
                $logger('worker:' . $worker->pid . ' redis error');
                sleep(1);
            }
        }
        $logger('worker:' . $worker->pid . ' exit');
    }

}